package tkM

import (
	"fmt"
	"db2s/Er"
	metaD "db2s/MetaInit"
	"db2s/arg"
	"db2s/global"
	"db2s/go-log/log"
	"db2s/outPut"
	"db2s/parDef"
	"db2s/ref"
	mq "db2s/topic-mq"
	"reflect"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

// table sum pod Init
func (t CheckSchemaMetaData) podInit(seq int, v arg.InputTableAttributes) (result *outPut.TPod) {
	var status int64
	result = &outPut.TPod{Seq: seq, Schema: v.Schema, Table: v.Table, Options: "none", Role: v.TableAttributes.Role,
		CheckMode: GlobalPConfigs.rules.CheckMode, IndexName: "missIndex", IndexType: "none", IndexCol: "none", Status: &status, Result: "ok"}
	switch GlobalPConfigs.rules.CheckMode {
	case "task":
		var sum int64
		result.Task = outPut.Task{
			Sum:            &sum,
			RateEvaluation: "差",
			Histogram:      "0.00",
		}
	}
	return
}

// writeOptimizer 目标端写入的相关优化
func writeOptimizer(plan2 *SchedulePlan2) {
	var (
		thread int64 = 1
		event        = "[writeOptimizer]"
		t      Er.TablesMetaInfoEr
		err    error
		avg    global.AvgRowLengthReturnResult
	)
	if t, err = Er.MetaInfo(Er.TableInfoMeta{DBType: GlobalPConfigs.dSns.SrcDBName}); err != nil {
		log.ErrorLog().Error(fmt.Sprintf("(%v) %v dest write optimizer fail. error is {%v}", plan2.Object.ObjectLogSeq, event, err))
		return
	}
	avg, err = t.AvgRowLength(
		global.TablesMetaInfoInput{
			LogSeq: 1,
			//DB:     GlobalPConfigs.SDB,
			TableInfo: global.TableInfo{
				Schema:           plan2.Object.Schema,
				Table:            plan2.Object.Table,
				BackendTableName: plan2.Object.Table,
			},
		})
	if err != nil {
		log.MainLog().Warn(fmt.Sprintf("(%v) %v dest write optimizer fail. error is {%v}. it will use default value.", plan2.Object.ObjectLogSeq, event, err))
		plan2.writeOptimizer.FixSum = 10
		plan2.ReadOptimizer.ChunkSum = 10
	}
	var b int64
	if avg.Result > 0 {
		b = 16 * 1024 * 64 / avg.Result
	}
	if b > 0 {
		plan2.writeOptimizer.FixSum = b
		plan2.ReadOptimizer.ChunkSum = b
		log.MainLog().Info(fmt.Sprintf("(%v) %v dest write optimizer chunkSum is {%v},avg length result is {%v}", plan2.Object.ObjectLogSeq, event, b, avg.Result))
	} else {
		plan2.writeOptimizer.FixSum = 10
		plan2.ReadOptimizer.ChunkSum = 10
		log.MainLog().Warn(fmt.Sprintf("(%v) %v dest chunk sum result is 0.avg length result is {%v}, it will use default value.", plan2.Object.ObjectLogSeq, event, avg.Result))
	}
	plan2.writeOptimizer.Thread = &thread
}

// readOptimizer 目标端写入的相关优化
func readOptimizer(IndexData metaD.IndexChooseResult, plan2 *SchedulePlan2) {
	var thread int64 = 1
	plan2.ReadOptimizer.IndexColumnType = "missIndex"
	plan2.ReadOptimizer.IndexColumn = IndexData.IndexColumn
	plan2.ReadOptimizer.IndexCardinality = IndexData.Cardinality
	plan2.ReadOptimizer.IndexColumnType = IndexData.IndexType
	plan2.ReadOptimizer.IndexColumnSeq = IndexData.ColumnSeq
	plan2.ReadOptimizer.IndexName = IndexData.IndexName
	plan2.TPod.IndexName = func() string {
		if len(IndexData.IndexName) > 0 {
			return IndexData.IndexName
		}
		return "missIndex"
	}()
	plan2.TPod.IndexType = func() string {
		if len(IndexData.IndexType) > 0 {
			return IndexData.IndexType
		}
		return "none"
	}()
	plan2.ReadOptimizer.ChunkStartSource = "src"
	plan2.ReadOptimizer.Thread = &thread
}

func (sp *SchedulePlan2) tableMetaArr() bool {
	var (
		event   = "[table Meta Check]"
		groupBy int
	)
	for _, v := range sp.Object.TableColData.SColumnInfo {
		if !GlobalPConfigs.rules.Task.IgColumn {
			var status bool
			for _, l := range sp.Object.TableColData.DColumnInfo {
				if strings.EqualFold(v.ColumnName, l.ColumnName) {
					status = true
					break
				}
			}
			if !status {
				log.ErrorLog().Error(fmt.Sprintf("%v There are differences in the structural data of table %v.%v on the original target side. src data is {%v}", event, sp.Object.Schema, sp.Object.Table, v.ColumnName))
				return false
			}
		} else {
			for _, v1 := range sp.Object.TableColData.DColumnInfo {
				if strings.EqualFold(v.ColumnName, v1.ColumnName) {
					groupBy++
					break
				}
			}
		}
	}
	if groupBy == 0 {
		return false
	}
	return true
}
func tableStatusInit(plan2 *SchedulePlan2) {
	var (
		errEnd int64 = 0
		logSeq int64
	)
	plan2.Status.LogSeq = logSeq
	plan2.Status.ErrEnd = &errEnd
}
func initXlsResult(s *SchemaInitContext) *outPut.WorkSheetResult {
	return &outPut.WorkSheetResult{
		Data: outPut.SheetResultData{
			Seq:    s.Seq,
			Schema: s.Schema,
			Table:  s.Table,
			Role:   s.ObjectRole,
			Task:   outPut.TaskResultSheet{},
		},
	}
}
func initXlsMiss(s *SchemaInitContext) *outPut.WorkSheetMiss {
	return &outPut.WorkSheetMiss{
		Data: outPut.SheetMiss{
			Schema: s.Schema,
			Table:  s.Table,
			Role:   s.ObjectRole,
		},
	}
}
func initXlsTerminal(_ *SchemaInitContext) *outPut.WorkSheetTerminal {
	return &outPut.WorkSheetTerminal{
		Data: outPut.TPod{},
	}
}
func initBaseSchedulePlanObject(s *SchemaInitContext) global.Object {
	return global.Object{
		TaskSeq: s.Seq, ObjectLogSeq: s.ObjectLogSeq, Schema: s.Schema, Table: s.Table, MSchema: s.MSchema, MTable: s.MTable,
		Role: s.ObjectRole,
		Input: global.StartPartConfigInputP{SqlMode: GlobalPConfigs.rules.Task.SqlMode,
			Scn:             GlobalPConfigs.rules.Task.Scn,
			SqlExecStopTime: time.Second * time.Duration(GlobalPConfigs.rules.Task.TimeOut),
		}}
}
func optionsInit(s *SchemaInitContext) parDef.Options {
	return parDef.Options{
		Scn: GlobalPConfigs.rules.Task.Scn,
		//StopTime: time.Duration(GlobalPConfigs.rules.Sync.),
		//WhereSql:     s.TableAttributes.WhereSql,
		//WhereAdd:     s.TableAttributes.WhereAdd,
		SqlMode: GlobalPConfigs.rules.Task.SqlMode,
		//SqlLogBin:    GlobalPConfigs.rules.Task.SqlLogBin,
		//RowsLimit:    GlobalPConfigs.rules.Task.RowsLimit,
		RepairMethod: GlobalPConfigs.repair.DataFix,
	}
}
func srcParameterInit(s *SchemaInitContext) parDef.Parameter {
	return parDef.Parameter{
		Con:     GlobalPConfigs.db.Source.GetDB("single"),
		Object:  parDef.Object{Schema: s.Schema, Table: s.Table},
		Options: optionsInit(s),
	}
}
func dstParameterInit(s *SchemaInitContext) parDef.Parameter {
	return parDef.Parameter{
		Con:     GlobalPConfigs.db.Target.GetDB("single"),
		Object:  parDef.Object{Schema: s.MSchema, Table: s.MTable},
		Options: optionsInit(s),
	}
}
func newSyncSchedulePlan(s *SchemaInitContext) *SchedulePlan2 {
	//var sum int64 = 0
	plan2 := &SchedulePlan2{
		Object:       initBaseSchedulePlanObject(s),
		SrcParameter: srcParameterInit(s),
		DstParameter: dstParameterInit(s),
		XlsResult:    initXlsResult(s),
		XlsMiss:      initXlsMiss(s),
		XlsTerminal:  initXlsTerminal(s),
		TPod:         s.TPod,
	}
	tableStatusInit(plan2)
	return plan2
}
func speedAdd(event string, plan1 any) (res any) {
	plan := plan1.(*SchedulePlan2)
	if atomic.LoadInt64(plan.ReadOptimizer.Thread) > 5 {
		atomic.SwapInt64(plan.ReadOptimizer.Thread, atomic.LoadInt64(plan.ReadOptimizer.Thread)-1)
	}
	if atomic.LoadInt64(plan.writeOptimizer.Thread) > 1 {
		atomic.SwapInt64(plan.writeOptimizer.Thread, atomic.LoadInt64(plan.writeOptimizer.Thread)-1)
	}
	return
}
func speedDel(event string, plan1 any) (res any) {
	plan := plan1.(*SchedulePlan2)
	atomic.SwapInt64(plan.ReadOptimizer.Thread, atomic.LoadInt64(plan.ReadOptimizer.Thread)+1)
	atomic.SwapInt64(plan.writeOptimizer.Thread, atomic.LoadInt64(plan.writeOptimizer.Thread)+1)
	return
}
func (plan *SchedulePlan2) speedResSend(event string, plan1 any) (res any) {
	if GlobalPConfigs.result.Teletypewriter == "bar" {
		if PlanContext.TaskBarSubsStatusGet(plan.subTaskInfo) != -2 {
			PlanContext.SpeedTaskBarAccumulate(plan1)
		}
	}
	return
}
func speedInit(plan *SchedulePlan2) mq.SpeedLimit {
	q := mq.NewSpeedMonitor(plan.Object.Schema, plan.Object.Table, plan.ReadOptimizer.ChunkSum)
	q.SpeedAddFunc = global.EFunc{FuncName: speedAdd, Params: plan}
	q.SpeedDelFunc = global.EFunc{FuncName: speedDel, Params: plan}
	q.SendResFunc = global.EFunc{FuncName: plan.speedResSend}
	return q
}
func initSendMsg(plan *SchedulePlan2) {
	plan.MQ.SendMsg = mq.ChanSendMsg{
		Table:             plan.Object.Table,
		MsgData:           plan.MQ.SendMsg.BaseMsg,
		MsgSuccessfulFunc: SendMsgExecSumFunc([]any{plan.MQ.FirstMsg.ProductMsg, int64(1)}),
		MsgFinishFunc:     SendMsgExecSumFunc([]any{[]any{plan.MQ.FirstMsg.CustomerObject}, int64(1)}),
		MsgProducer:       producer,
	}
}

func initTaskSchedulePlan(s *SchemaInitContext) (plan2 *SchedulePlan2) {
	plan2 = newSyncSchedulePlan(s)
	readOptimizer(s.IndexData, plan2)
	//writeOptimizer(plan2)
	plan2.MQ = mq.NewMonitorMsgMq()
	plan2.MQ.Speed = speedInit(plan2)
	initSendMsg(plan2)
	subTaskInit(plan2)
	if s.TPod.TableMiss {
		plan2.Object.TableMiss = true
		return
	}
	return
}
func InitToSchedulePlan2TableExecFunc(event string, f any) (r any) {
	_, chanParameter, userParameter := schemaMetaExecFuncParameterSplit(f.([]any))
	ChanSchedulePlanTable := userParameter[0].(chan any)
	if funcName, ok := ScheduleTaskInitFuncRef[GlobalPConfigs.rules.CheckMode]; ok {
		if funcName.Kind() == reflect.Func {
			arguments := []reflect.Value{
				reflect.ValueOf(chanParameter)}
			m1 := funcName.Call(arguments)
			var m2 *SchedulePlan2
			for i := 0; i < len(m1); i++ {
				m2 = m1[i].Interface().(*SchedulePlan2)
			}
			ChanSchedulePlanTable <- m2
		}
	}
	return
}
func InitToSchedulePlan2TableFinishFunc(event string, _ any) (r any) {
	ref.MyWaitDel("scheduleObject")
	return
}
func plan2TableScheduleObjectInit(f2 *SchemaInitContext, ChanSchedulePlanTable any) mq.MonitorMsgMq {
	ref.MyWaitAdd("scheduleObject")
	var wg sync.WaitGroup
	return mq.NewMonitorMsgMq1().SetSendMsg(f2.ChanSchemaContext).SetCurryLimit(mq.SendCurryLimit{
		Switch:   true,
		Wg:       &wg,
		CurrySum: make(chan struct{}, GlobalPConfigs.rules.Mtc),
	}).SetNormalQuitFunc(global.EFunc{FuncName: InitToSchedulePlan2TableFinishFunc}).SetExecFunc([]global.EFunc{{FuncName: InitToSchedulePlan2TableExecFunc, Params: []any{ChanSchedulePlanTable}}})
	//send = mq.NewMonitorMsgMq()
	//send.SendMsg.MsgModifyKey = f2.ChanSchemaContext
	//send.NormalQuit = global.EFunc{FuncName: InitToSchedulePlan2TableFinishFunc}
	//send.ExecFunc = []global.EFunc{{FuncName: InitToSchedulePlan2TableExecFunc, Params: []any{ChanSchedulePlanTable}}}
	//return send
}

// InitToSchedulePlan2Table 初始化 执行计划 以 map 的形式 key 表名 value SchedulePlan2
func InitToSchedulePlan2Table() (ChanSchedulePlanTable chan any) {
	var f2 = global.GetIoc().GetBean("schemaInitContext").(*SchemaInitContext)
	ChanSchedulePlanTable = make(chan any, f2.TableSum)
	go func() {
		defer func() {
			close(ChanSchedulePlanTable)
		}()
		plan2TableScheduleObjectInit(f2, ChanSchedulePlanTable).NormalChanSendMsg("[taskInitToSchedulePlan2Table]")
	}()
	return
}
